🎖️GitЯра🎖️
Commit e48640e4734003cebc9292d7a45400a674f1612a
Parents : 4e48e64
Author : James Rich <2199651+jamesarich@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-07-11T19:12:57-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-07-12T00:12:57Z
fix(database): route all one-shot DB writes through the merge drain barrier (#6233)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Changes
7 files changed, 234 insertions(+), 144 deletions(-)
Diff
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt
index dfc214ead5..891ed6efc1 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/datasource/EventFirmwareEditionLocalDataSource.kt
@@ -27,20 +27,23 @@ class EventFirmwareEditionLocalDataSource(
private val dbManager: DatabaseProvider,
private val dispatchers: CoroutineDispatchers,
) {
+ // Reads may use the direct accessor; writes go through withDb so they register with the cross-transport merge
+ // drain barrier (see DatabaseProvider).
private val dao
get() = dbManager.currentDb.value.eventFirmwareEditionDao()
suspend fun getByEdition(edition: String): EventFirmwareEditionEntity? =
withContext(dispatchers.io) { dao.getByEdition(edition) }
- suspend fun upsertAll(editions: List<EventFirmwareEditionEntity>) =
- withContext(dispatchers.io) { dao.upsertAll(editions) }
+ suspend fun upsertAll(editions: List<EventFirmwareEditionEntity>) {
+ withContext(dispatchers.io) { dbManager.withDb { it.eventFirmwareEditionDao().upsertAll(editions) } }
+ }
// No-op on empty: `NOT IN ()` is always true in SQLite, so forwarding an empty list would wipe the whole table.
// Callers only prune against a non-empty upstream payload; an empty list means "keep everything".
suspend fun deleteNotIn(keep: List<String>) {
if (keep.isEmpty()) return
- withContext(dispatchers.io) { dao.deleteNotIn(keep) }
+ withContext(dispatchers.io) { dbManager.withDb { it.eventFirmwareEditionDao().deleteNotIn(keep) } }
}
suspend fun count(): Int = withContext(dispatchers.io) { dao.count() }
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt
index 57e6312b0c..52698b5264 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/MeshLogRepositoryImpl.kt
@@ -153,51 +153,66 @@ open class MeshLogRepositoryImpl(
.mapLatest { list -> list.map { it.asExternalModel() }.firstOrNull { it.myNodeInfo != null }?.myNodeInfo }
.flowOn(dispatchers.io)
+ // Writes go through withDb so they register with the cross-transport merge drain barrier (see DatabaseProvider).
+
/** Persists a new log entry to the database if logging is enabled in preferences. */
override suspend fun insert(log: MeshLog) = withContext(dispatchers.io) {
if (!meshLogPrefs.loggingEnabled.value) return@withContext
- dbManager.currentDb.value.meshLogDao().insert(log.asEntity())
+ dbManager.withDb { it.meshLogDao().insert(log.asEntity()) }
+ Unit
}
/** Clears all logs from the database. */
- override suspend fun deleteAll() =
- withContext(dispatchers.io) { dbManager.currentDb.value.meshLogDao().deleteAll() }
+ override suspend fun deleteAll() {
+ withContext(dispatchers.io) { dbManager.withDb { it.meshLogDao().deleteAll() } }
+ }
/** Deletes a specific log entry by its [uuid]. */
- override suspend fun deleteLog(uuid: String) =
- withContext(dispatchers.io) { dbManager.currentDb.value.meshLogDao().deleteLog(uuid) }
+ override suspend fun deleteLog(uuid: String) {
+ withContext(dispatchers.io) { dbManager.withDb { it.meshLogDao().deleteLog(uuid) } }
+ }
/** Deletes all logs associated with a specific [nodeNum] and [portNum]. */
override suspend fun deleteLogs(nodeNum: Int, portNum: Int) = withContext(dispatchers.io) {
val myNodeNum = nodeInfoReadDataSource.myNodeInfoFlow().firstOrNull()?.myNodeNum
val logId = if (nodeNum == myNodeNum) MeshLog.NODE_NUM_LOCAL else nodeNum
- dbManager.currentDb.value.meshLogDao().deleteLogs(logId, portNum)
+ dbManager.withDb { it.meshLogDao().deleteLogs(logId, portNum) }
+ Unit
}
/** Deletes only local stats telemetry logs for [nodeNum], preserving other telemetry types. */
override suspend fun deleteLocalStatsLogs(nodeNum: Int) = withContext(dispatchers.io) {
val myNodeNum = nodeInfoReadDataSource.myNodeInfoFlow().firstOrNull()?.myNodeNum
val logId = if (nodeNum == myNodeNum) MeshLog.NODE_NUM_LOCAL else nodeNum
- val dao = dbManager.currentDb.value.meshLogDao()
- val localStatsLogs =
- dao.getLogsFrom(logId, PortNum.TELEMETRY_APP.value, Int.MAX_VALUE)
- .firstOrNull()
- .orEmpty()
- .map { it.asExternalModel() }
- .filter { parseTelemetryLog(it)?.local_stats != null }
-
- val localStatsLogIds = localStatsLogs.map { it.uuid }
- // Chunk to stay under SQLite's bind-variable limit; re-fetch DAO per chunk if the active DB switches.
- for (chunk in localStatsLogIds.chunked(DELETE_CHUNK_SIZE)) {
- dbManager.currentDb.value.meshLogDao().deleteLogsByUuid(chunk)
+ // One withDb block: the query and the chunked deletes all land on the same DB instance (previously each
+ // chunk
+ // re-fetched the DAO to survive a mid-op switch; capturing one instance is the more correct behavior, and
+ // the
+ // drain barrier now covers the whole read-modify-delete against a concurrent merge).
+ dbManager.withDb { db ->
+ val dao = db.meshLogDao()
+ val localStatsLogIds =
+ dao.getLogsFrom(logId, PortNum.TELEMETRY_APP.value, Int.MAX_VALUE)
+ .firstOrNull()
+ .orEmpty()
+ .map { it.asExternalModel() }
+ .filter { parseTelemetryLog(it)?.local_stats != null }
+ .map { it.uuid }
+
+ // Chunk to stay under SQLite's bind-variable limit.
+ for (chunk in localStatsLogIds.chunked(DELETE_CHUNK_SIZE)) {
+ dao.deleteLogsByUuid(chunk)
+ }
}
+ Unit
}
/** Prunes the log database based on the configured [retentionDays]. */
@Suppress("MagicNumber")
override suspend fun deleteLogsOlderThan(retentionDays: Int) = withContext(dispatchers.io) {
val cutoffTime = nowMillis - (retentionDays.toLong() * 24 * 60 * 60 * 1000)
- dbManager.currentDb.value.meshLogDao().deleteOlderThan(cutoffTime)
+ dbManager.withDb { it.meshLogDao().deleteOlderThan(cutoffTime) }
+ Unit
}
companion object {
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt
index 8ea396944b..0f42fef0aa 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt
@@ -31,6 +31,7 @@ import okio.ByteString.Companion.toByteString
import org.koin.core.annotation.Single
import org.meshtastic.core.database.DatabaseProvider
import org.meshtastic.core.database.dao.NodeInfoDao
+import org.meshtastic.core.database.dao.PacketDao
import org.meshtastic.core.database.entity.PacketEntity
import org.meshtastic.core.database.entity.toReaction
import org.meshtastic.core.di.CoroutineDispatchers
@@ -91,36 +92,46 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
override fun getUnreadCountTotal(): Flow<Int> =
dbManager.currentDb.flatMapLatest { db -> db.packetDao().getUnreadCountTotal() }
- override suspend fun clearUnreadCount(contact: String, timestamp: Long) =
+ // One-shot writes go through withDb so they register with the cross-transport merge drain barrier and pick up
+ // its closed-pool retry (see DatabaseProvider). Reads and Flow/Paging factories stay on currentDb by design.
+
+ override suspend fun clearUnreadCount(contact: String, timestamp: Long) {
withContext(dispatchers.io + NonCancellable) {
- dbManager.currentDb.value.packetDao().clearUnreadCount(contact, timestamp)
+ dbManager.withDb { it.packetDao().clearUnreadCount(contact, timestamp) }
}
+ }
- override suspend fun clearAllUnreadCounts() =
- withContext(dispatchers.io + NonCancellable) { dbManager.currentDb.value.packetDao().clearAllUnreadCounts() }
+ override suspend fun clearAllUnreadCounts() {
+ withContext(dispatchers.io + NonCancellable) { dbManager.withDb { it.packetDao().clearAllUnreadCounts() } }
+ }
- override suspend fun updateLastReadMessage(contact: String, messageUuid: Long, lastReadTimestamp: Long) =
+ override suspend fun updateLastReadMessage(contact: String, messageUuid: Long, lastReadTimestamp: Long) {
withContext(dispatchers.io + NonCancellable) {
- val dao = dbManager.currentDb.value.packetDao()
- val current = dao.getContactSettings(contact)
- val existingTimestamp = current?.lastReadMessageTimestamp ?: Long.MIN_VALUE
- if (lastReadTimestamp <= existingTimestamp) {
- return@withContext
+ // Read-modify-write in one withDb block so both statements land on the same DB instance.
+ dbManager.withDb { db ->
+ val dao = db.packetDao()
+ val current = dao.getContactSettings(contact)
+ val existingTimestamp = current?.lastReadMessageTimestamp ?: Long.MIN_VALUE
+ if (lastReadTimestamp <= existingTimestamp) {
+ return@withDb
+ }
+ val updated =
+ (current ?: ContactSettingsEntity(contact_key = contact)).copy(
+ lastReadMessageUuid = messageUuid,
+ lastReadMessageTimestamp = lastReadTimestamp,
+ )
+ dao.upsertContactSettings(listOf(updated))
}
- val updated =
- (current ?: ContactSettingsEntity(contact_key = contact)).copy(
- lastReadMessageUuid = messageUuid,
- lastReadMessageTimestamp = lastReadTimestamp,
- )
- dao.upsertContactSettings(listOf(updated))
}
+ }
override suspend fun getQueuedPackets(): List<DataPacket> = withContext(dispatchers.io) {
dbManager.currentDb.value.packetDao().getAllDataPackets().filter { it.status == MessageStatus.QUEUED }
}
- suspend fun insertRoomPacket(packet: RoomPacket) =
- withContext(dispatchers.io + NonCancellable) { dbManager.currentDb.value.packetDao().insert(packet) }
+ suspend fun insertRoomPacket(packet: RoomPacket) {
+ withContext(dispatchers.io + NonCancellable) { dbManager.withDb { it.packetDao().insert(packet) } }
+ }
override suspend fun savePacket(
myNodeNum: Int,
@@ -230,17 +241,21 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
}
}
- override suspend fun updateMessageStatus(d: DataPacket, m: MessageStatus) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().updateMessageStatus(d, m) }
+ override suspend fun updateMessageStatus(d: DataPacket, m: MessageStatus) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().updateMessageStatus(d, m) } }
+ }
- override suspend fun updateMessageId(d: DataPacket, id: Int) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().updateMessageId(d, id) }
+ override suspend fun updateMessageId(d: DataPacket, id: Int) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().updateMessageId(d, id) } }
+ }
- override suspend fun setMessageTranslation(uuid: Long, translatedText: String) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setTranslation(uuid, translatedText) }
+ override suspend fun setMessageTranslation(uuid: Long, translatedText: String) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().setTranslation(uuid, translatedText) } }
+ }
- override suspend fun setShowTranslated(uuid: Long, showTranslated: Boolean) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setShowTranslated(uuid, showTranslated) }
+ override suspend fun setShowTranslated(uuid: Long, showTranslated: Boolean) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().setShowTranslated(uuid, showTranslated) } }
+ }
override suspend fun getPacketById(id: Int): DataPacket? =
withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().getPacketById(id)?.data }
@@ -296,29 +311,38 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
}
override suspend fun update(packet: DataPacket, routingError: Int): Unit = withContext(dispatchers.io) {
- val dao = dbManager.currentDb.value.packetDao()
- // Match on key fields that identify the packet, rather than the entire data object
- dao.findPacketsWithId(packet.id)
- .find { it.data.id == packet.id && it.data.from == packet.from && it.data.to == packet.to }
- ?.let { existing ->
- val updated =
- if (routingError >= 0) {
- existing.copy(data = packet, routingError = routingError)
- } else {
- existing.copy(data = packet)
- }
- dao.update(updated)
- }
+ // Read-modify-write in one withDb block so the lookup and update land on the same DB instance.
+ dbManager.withDb { db ->
+ val dao = db.packetDao()
+ // Match on key fields that identify the packet, rather than the entire data object
+ dao.findPacketsWithId(packet.id)
+ .find { it.data.id == packet.id && it.data.from == packet.from && it.data.to == packet.to }
+ ?.let { existing ->
+ val updated =
+ if (routingError >= 0) {
+ existing.copy(data = packet, routingError = routingError)
+ } else {
+ existing.copy(data = packet)
+ }
+ dao.update(updated)
+ }
+ }
+ Unit
}
- override suspend fun insertReaction(reaction: Reaction, myNodeNum: Int) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().insert(reaction.toEntity(myNodeNum)) }
+ override suspend fun insertReaction(reaction: Reaction, myNodeNum: Int) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().insert(reaction.toEntity(myNodeNum)) } }
+ }
- override suspend fun updateReaction(reaction: Reaction) = withContext(dispatchers.io) {
- val dao = dbManager.currentDb.value.packetDao()
- dao.findReactionsWithId(reaction.packetId)
- .find { it.userId == reaction.user.id && it.emoji == reaction.emoji }
- ?.let { dao.update(reaction.toEntity(it.myNodeNum)) } ?: Unit
+ override suspend fun updateReaction(reaction: Reaction) {
+ withContext(dispatchers.io) {
+ dbManager.withDb { db ->
+ val dao = db.packetDao()
+ dao.findReactionsWithId(reaction.packetId)
+ .find { it.userId == reaction.user.id && it.emoji == reaction.emoji }
+ ?.let { dao.update(reaction.toEntity(it.myNodeNum)) }
+ }
+ }
}
override suspend fun getReactionByPacketId(packetId: Int): Reaction? = withContext(dispatchers.io) {
@@ -329,17 +353,10 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
dbManager.currentDb.value.packetDao().findPacketsWithId(packetId).map { it.data }
}
- private suspend fun findPacketsWithIdInternal(packetId: Int) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().findPacketsWithId(packetId) }
-
override suspend fun findReactionsWithId(packetId: Int): List<Reaction> = withContext(dispatchers.io) {
dbManager.currentDb.value.packetDao().findReactionsWithId(packetId).toReaction { null }
}
- private suspend fun findReactionsWithIdInternal(packetId: Int) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().findReactionsWithId(packetId) }
-
- @Suppress("CyclomaticComplexMethod")
override suspend fun updateSFPPStatus(
packetId: Int,
from: Int,
@@ -349,9 +366,28 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
rxTime: Long,
myNodeNum: Int?,
) = withContext(dispatchers.io) {
- val dao = dbManager.currentDb.value.packetDao()
- val packets = findPacketsWithIdInternal(packetId)
- val reactions = findReactionsWithIdInternal(packetId)
+ // One withDb block: the id lookups and the status updates all land on the same DB instance.
+ dbManager.withDb { db ->
+ val dao = db.packetDao()
+ val packets = dao.findPacketsWithId(packetId)
+ val reactions = dao.findReactionsWithId(packetId)
+ updateSFPPMatches(dao, packets, reactions, from, to, hash, status, rxTime, myNodeNum)
+ }
+ Unit
+ }
+
+ @Suppress("CyclomaticComplexMethod", "LongParameterList")
+ private suspend fun updateSFPPMatches(
+ dao: PacketDao,
+ packets: List<RoomPacket>,
+ reactions: List<RoomReaction>,
+ from: Int,
+ to: Int,
+ hash: ByteArray,
+ status: MessageStatus,
+ rxTime: Long,
+ myNodeNum: Int?,
+ ) {
val fromId = NodeAddress.numToDefaultId(from)
val isFromLocalNode = myNodeNum != null && from == myNodeNum
val toId =
@@ -401,8 +437,7 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
return@forEach
}
val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else reaction.timestamp
- val updatedReaction =
- reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
+ val updatedReaction = reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
dao.update(updatedReaction)
}
}
@@ -410,46 +445,59 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
override suspend fun updateSFPPStatusByHash(hash: ByteArray, status: MessageStatus, rxTime: Long): Unit =
withContext(dispatchers.io) {
- val dao = dbManager.currentDb.value.packetDao()
- val hashByteString = hash.toByteString()
- dao.findPacketBySfppHash(hashByteString)?.let { packet ->
- // If it's already confirmed, don't downgrade it
- if (packet.data.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
- return@let
+ // Read-modify-write in one withDb block so the hash lookups and updates land on the same DB instance.
+ dbManager.withDb { db ->
+ val dao = db.packetDao()
+ val hashByteString = hash.toByteString()
+ dao.findPacketBySfppHash(hashByteString)?.let { packet ->
+ // If it's already confirmed, don't downgrade it
+ if (packet.data.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
+ return@let
+ }
+ val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else packet.received_time
+ val updatedData = packet.data.copy(status = status, sfppHash = hashByteString, time = newTime)
+ dao.update(packet.copy(data = updatedData, sfpp_hash = hashByteString, received_time = newTime))
}
- val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else packet.received_time
- val updatedData = packet.data.copy(status = status, sfppHash = hashByteString, time = newTime)
- dao.update(packet.copy(data = updatedData, sfpp_hash = hashByteString, received_time = newTime))
- }
- dao.findReactionBySfppHash(hashByteString)?.let { reaction ->
- if (reaction.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
- return@let
+ dao.findReactionBySfppHash(hashByteString)?.let { reaction ->
+ if (reaction.status == MessageStatus.SFPP_CONFIRMED && status == MessageStatus.SFPP_ROUTING) {
+ return@let
+ }
+ val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else reaction.timestamp
+ val updatedReaction =
+ reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
+ dao.update(updatedReaction)
}
- val newTime = if (rxTime > 0) rxTime * MILLISECONDS_IN_SECOND else reaction.timestamp
- val updatedReaction = reaction.copy(status = status, sfpp_hash = hashByteString, timestamp = newTime)
- dao.update(updatedReaction)
}
+ Unit
}
override suspend fun deleteMessages(uuidList: List<Long>) = withContext(dispatchers.io) {
- for (chunk in uuidList.chunked(DELETE_CHUNK_SIZE)) {
- // Fetch DAO per chunk to avoid holding a stale reference if the active DB switches
- dbManager.currentDb.value.packetDao().deleteMessages(chunk)
+ // One withDb block: every chunk lands on the same DB instance (previously each chunk re-fetched the DAO to
+ // survive a mid-op switch; the drain barrier now covers the whole chunked delete instead).
+ dbManager.withDb { db ->
+ for (chunk in uuidList.chunked(DELETE_CHUNK_SIZE)) {
+ db.packetDao().deleteMessages(chunk)
+ }
}
+ Unit
}
- override suspend fun deleteContacts(contactList: List<String>) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteContacts(contactList) }
+ override suspend fun deleteContacts(contactList: List<String>) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().deleteContacts(contactList) } }
+ }
- override suspend fun deleteWaypoint(id: Int) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteWaypoint(id) }
+ override suspend fun deleteWaypoint(id: Int) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().deleteWaypoint(id) } }
+ }
- suspend fun delete(packet: RoomPacket) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().delete(packet) }
+ suspend fun delete(packet: RoomPacket) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().delete(packet) } }
+ }
- suspend fun update(packet: RoomPacket) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().update(packet) }
+ suspend fun update(packet: RoomPacket) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().update(packet) } }
+ }
override fun getContactSettings(): Flow<Map<String, ContactSettings>> = dbManager.currentDb
.flatMapLatest { db -> db.packetDao().getContactSettings() }
@@ -459,14 +507,17 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
dbManager.currentDb.value.packetDao().getContactSettings(contact)?.toShared() ?: ContactSettings(contact)
}
- override suspend fun setMuteUntil(contacts: List<String>, until: Long) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().setMuteUntil(contacts, until) }
+ override suspend fun setMuteUntil(contacts: List<String>, until: Long) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().setMuteUntil(contacts, until) } }
+ }
- suspend fun insertReaction(reaction: RoomReaction) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().insert(reaction) }
+ suspend fun insertReaction(reaction: RoomReaction) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().insert(reaction) } }
+ }
- suspend fun updateReaction(reaction: RoomReaction) =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().update(reaction) }
+ suspend fun updateReaction(reaction: RoomReaction) {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().update(reaction) } }
+ }
override fun getFilteredCountFlow(contactKey: String): Flow<Int> =
dbManager.currentDb.flatMapLatest { db -> db.packetDao().getFilteredCountFlow(contactKey) }
@@ -474,22 +525,25 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
override suspend fun getFilteredCount(contactKey: String): Int =
withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().getFilteredCount(contactKey) }
- override suspend fun setContactFilteringDisabled(contactKey: String, disabled: Boolean) =
+ override suspend fun setContactFilteringDisabled(contactKey: String, disabled: Boolean) {
withContext(dispatchers.io) {
- dbManager.currentDb.value.packetDao().setContactFilteringDisabled(contactKey, disabled)
+ dbManager.withDb { it.packetDao().setContactFilteringDisabled(contactKey, disabled) }
}
+ }
- override suspend fun clearPacketDB() =
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().deleteAll() }
+ override suspend fun clearPacketDB() {
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().deleteAll() } }
+ }
- override suspend fun migrateChannelsByPSK(oldSettings: List<ChannelSettings>, newSettings: List<ChannelSettings>) =
+ override suspend fun migrateChannelsByPSK(oldSettings: List<ChannelSettings>, newSettings: List<ChannelSettings>) {
withContext(dispatchers.io) {
- dbManager.currentDb.value.packetDao().migrateChannelsByPSK(oldSettings, newSettings)
+ dbManager.withDb { it.packetDao().migrateChannelsByPSK(oldSettings, newSettings) }
}
+ }
override suspend fun updateFilteredBySender(senderId: String, filtered: Boolean) {
val pattern = "%\"from\":\"${senderId}\"%"
- withContext(dispatchers.io) { dbManager.currentDb.value.packetDao().updateFilteredBySender(pattern, filtered) }
+ withContext(dispatchers.io) { dbManager.withDb { it.packetDao().updateFilteredBySender(pattern, filtered) } }
}
private fun org.meshtastic.core.database.dao.PacketDao.getAllWaypointsFlow(): Flow<List<RoomPacket>> =
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt
index 6546b6504b..2e0f994aaf 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/QuickChatActionRepositoryImpl.kt
@@ -34,21 +34,20 @@ class QuickChatActionRepositoryImpl(
override fun getAllActions(): Flow<List<QuickChatAction>> =
dbManager.currentDb.flatMapLatest { it.quickChatActionDao().getAll() }.flowOn(dispatchers.io)
+ // Writes go through withDb so they register with the cross-transport merge drain barrier (see DatabaseProvider).
override suspend fun upsert(action: QuickChatAction) {
- withContext(dispatchers.io) { dbManager.currentDb.value.quickChatActionDao().upsert(action) }
+ withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().upsert(action) } }
}
override suspend fun deleteAll() {
- withContext(dispatchers.io) { dbManager.currentDb.value.quickChatActionDao().deleteAll() }
+ withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().deleteAll() } }
}
override suspend fun delete(action: QuickChatAction) {
- withContext(dispatchers.io) { dbManager.currentDb.value.quickChatActionDao().delete(action) }
+ withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().delete(action) } }
}
override suspend fun setItemPosition(uuid: Long, newPos: Int) {
- withContext(dispatchers.io) {
- dbManager.currentDb.value.quickChatActionDao().updateActionPosition(uuid, newPos)
- }
+ withContext(dispatchers.io) { dbManager.withDb { it.quickChatActionDao().updateActionPosition(uuid, newPos) } }
}
}
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt
index 4ad7afc241..769d184795 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/TracerouteSnapshotRepositoryImpl.kt
@@ -43,20 +43,25 @@ class TracerouteSnapshotRepositoryImpl(
.flowOn(dispatchers.io)
.conflate()
- override suspend fun upsertSnapshotPositions(logUuid: String, requestId: Int, positions: Map<Int, Position>) =
+ // The delete+insert pair runs in one withDb block so both land on the same DB instance and stay visible to the
+ // cross-transport merge drain barrier (see DatabaseProvider).
+ override suspend fun upsertSnapshotPositions(logUuid: String, requestId: Int, positions: Map<Int, Position>) {
withContext(dispatchers.io) {
- val dao = dbManager.currentDb.value.tracerouteNodePositionDao()
- dao.deleteByLogUuid(logUuid)
- if (positions.isEmpty()) return@withContext
- val entities =
- positions.map { (nodeNum, position) ->
- TracerouteNodePositionEntity(
- logUuid = logUuid,
- requestId = requestId,
- nodeNum = nodeNum,
- position = position,
- )
- }
- dao.insertAll(entities)
+ dbManager.withDb {
+ val dao = it.tracerouteNodePositionDao()
+ dao.deleteByLogUuid(logUuid)
+ if (positions.isEmpty()) return@withDb
+ val entities =
+ positions.map { (nodeNum, position) ->
+ TracerouteNodePositionEntity(
+ logUuid = logUuid,
+ requestId = requestId,
+ nodeNum = nodeNum,
+ position = position,
+ )
+ }
+ dao.insertAll(entities)
+ }
}
+ }
}
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
index e98b40705a..885458a8c9 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
@@ -383,9 +383,15 @@ open class DatabaseManager(
}
}
- /** Atomically snapshots the active DB and registers a writer against it. Returns null if none is open. */
- private suspend fun beginWrite(): MeshtasticDatabase? = writerTrackerMutex.withLock {
- val db = _currentDb.value ?: return@withLock null
+ /**
+ * Atomically snapshots the active DB and registers a writer against it. Before the first [switchActiveDatabase]
+ * `_currentDb` is still null, so fall back to the public [currentDb] view — the eagerly-opened default DB — giving
+ * withDb the exact DB-resolution semantics of a direct `currentDb.value` caller. Desktop never calls [init] until a
+ * device is selected, so without the fallback pre-connection writes (quick-chat actions, firmware/hardware cache
+ * refreshes) would silently no-op there instead of landing in the default DB the pre-connection flows read from.
+ */
+ private suspend fun beginWrite(): MeshtasticDatabase = writerTrackerMutex.withLock {
+ val db = _currentDb.value ?: currentDb.value
activeWriters[db] = (activeWriters[db] ?: 0) + 1
db
}
@@ -433,7 +439,7 @@ open class DatabaseManager(
@Suppress("ReturnCount", "ThrowsCount", "TooGenericExceptionCaught", "CyclomaticComplexMethod")
private suspend fun <T> withCurrentDb(block: suspend (MeshtasticDatabase) -> T): T? {
- val db = beginWrite() ?: return null
+ val db = beginWrite()
val active = currentDbName
markLastUsed(active)
try {
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt
index e406ed7faa..fc9daabee0 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseProvider.kt
@@ -21,6 +21,14 @@ import kotlinx.coroutines.flow.StateFlow
/**
* Provides multiplatform access to the current [MeshtasticDatabase] and a safe transactional helper. Platform
* implementations manage the concrete lifecycle (Room on Android, etc.).
+ *
+ * **Write policy:** every one-shot DAO *write* (insert/upsert/update/delete/clear) must go through [withDb], never
+ * `currentDb.value` directly. [withDb] registers the write with the cross-transport merge drain barrier (see
+ * [DatabaseManager.associateDevice]) so a merge can't snapshot a database while the write is still in flight and lose
+ * it when that database is retired — and it retries once if the pool closes under a concurrent DB switch. Direct
+ * `currentDb.value` is fine for one-shot *reads* (a torn read against a just-retired DB is recoverable and reads don't
+ * need drain visibility), and `currentDb` itself is the right latch for Flow/Paging factories, which must re-latch on
+ * DB switch and must stay out of [withDb]'s containment lane.
*/
interface DatabaseProvider {
/** Reactive stream of the currently active database instance. */
Served by rngit 1.5.2 - Generated in 0.32s